package com.bd08.stock.connector;

import org.apache.kafka.common.config.ConfigDef;
import org.apache.kafka.connect.connector.Task;
import org.apache.kafka.connect.errors.ConnectException;
import org.apache.kafka.connect.source.SourceConnector;
import org.apache.kafka.common.config.ConfigDef.Importance;
import org.apache.kafka.common.config.ConfigDef.Type;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @author seal
 * @date 2022-03-11
 *
 * 这是采用kafka connector 规范编写的 拉取 股票数据的类
 * SourceConnector 规范 要实现 独有的方法
 */
public class FinanceSourceConnector extends SourceConnector {

    private static final Logger log = LoggerFactory.getLogger(FinanceSourceConnector.class);

    public static final String TOPIC_CONFIG ="topic";
    public static final String STOCK_SYMBOLS_CONFIG = "symbols";
    public static final String STOCK_PORTFOLIO_CONFIG = "portfolio";
    public static final String REFRESH_INTERVAL_CONFIG = "interval";
    public static final String SCHEMA_URI_CONFIG = "schema.registry.uri";
    public static final String SCHEMA_SUBJECT_CONFIG = "schema.subject";
    public static final String SPOOF_FLAG_CONFIG = "spoof";
    public static final String CUID = "cuid";


//public static final String TOPIC_CONFIG = "inputgupiao";//主题
public static final String TOPIC_CONFIG_DOC = "stock-gupiao";//要将数据发布到的主题
    public static final String TOPIC_CONFIG_DEFAULT = "stock";//股票
    //public static final String STOCK_SYMBOLS_CONFIG = "symbols";//库存_符号_配置
    public static final String STOCK_SYMBOLS_CONFIG_DOC = "List of stock symbols to process.";//要处理的股票符号列表。
    public static final String STOCK_SYMBOLS_CONFIG_DEFAULT = "INTC,TSLA";//库存符号配置默认值
    //public static final String STOCK_PORTFOLIO_CONFIG = "portfolio";//股票组合配置
    public static final String STOCK_PORTFOLIO_CONFIG_DOC = "list of predefined symbols";//股票投资组合配置文件
    public static final String STOCK_PORTFOLIO_CONFIG_DEFAULT = "none";//股票组合配置默认值
    //public static final String REFRESH_INTERVAL_CONFIG = "10";//刷新配置间隔
    public static final String REFRESH_INTERVAL_CONFIG_DOC = "30";//几秒钟内检查库存更新的频率
    public static final String REFRESH_INTERVAL_CONFIG_DEFAULT = "20";//刷新配置间隔的默认值
    //测试中未使用用Schema Registry
    //public static final String SCHEMA_URI_CONFIG = "schema.registry.uri";//模式 注册处 uri
    public static final String SCHEMA_URI_CONFIG_DOC = "The URI to the Schema Registry.";//架构注册表的URI
    public static final String SCHEMA_URI_CONFIG_DEFAULT = "http://localhost:8081";
    //public static final String SCHEMA_SUBJECT_CONFIG = "schema.subject";//模式 主题
    public static final String SCHEMA_SUBJECT_CONFIG_DOC = "The subject used to validate avro schema.";//用于验证的主题将具有模式
    public static final String SCHEMA_SUBJECT_CONFIG_DEFAULT = "n/a";

    //public static final String SPOOF_FLAG_CONFIG = "spoof";//欺骗标志配置
    public static final String SPOOF_FLAG_CONFIG_DOC = "RAND";//如果数据来自过去数据（PAST）或随机数据（RAND）或无数据NONE
    public static final String SPOOF_FLAG_CONFIG_DEFAULT = "RAND";//欺骗标志配置默认值
    //public static final String CUID = "cuid";//部分
    public static final String CUID_DOC = "cuid";//部分文件
    public static final String CUID_DEFAULT = "n/a";//部分默认值



    //配置的定义
    private static final ConfigDef CONFIG_DEF = new ConfigDef()
            //定义    主题配置     ConfigDef.Type type      主题_配置_默认值                 主题_配置_文档
            .define(TOPIC_CONFIG, Type.STRING, TOPIC_CONFIG_DEFAULT, Importance.HIGH, TOPIC_CONFIG_DOC)
            //定义    库存_符号_配置                        库存符号配置默认值                                   库存符号配置文件
            .define(STOCK_SYMBOLS_CONFIG, Type.STRING, STOCK_SYMBOLS_CONFIG_DEFAULT, Importance.HIGH, STOCK_SYMBOLS_CONFIG_DOC)
            //定义    欺骗标志配置                      欺骗标志配置默认值                               欺骗标志配置文件
            .define(SPOOF_FLAG_CONFIG, Type.STRING, SPOOF_FLAG_CONFIG_DEFAULT, Importance.HIGH, SPOOF_FLAG_CONFIG_DOC)
            //定义    股票组合配置                              股票组合配置默认值                                   股票投资组合配置文件
            .define(STOCK_PORTFOLIO_CONFIG, Type.STRING, STOCK_PORTFOLIO_CONFIG_DEFAULT, Importance.MEDIUM, STOCK_PORTFOLIO_CONFIG_DOC)
            //定义    刷新间隔配置                              刷新间隔配置默认值                                   刷新间隔配置文档
            .define(REFRESH_INTERVAL_CONFIG, Type.STRING, REFRESH_INTERVAL_CONFIG_DEFAULT, Importance.MEDIUM, REFRESH_INTERVAL_CONFIG_DOC)
            //定义                                    模式 URI配置默认值                             模式URI 配置文档
            .define(SCHEMA_URI_CONFIG, Type.STRING, SCHEMA_URI_CONFIG_DEFAULT, Importance.HIGH, SCHEMA_URI_CONFIG_DOC)
            //定义    模式_主题_配置                        模式 主题 配置 默认值                                模式主题配置文档
            .define(SCHEMA_SUBJECT_CONFIG, Type.STRING, SCHEMA_SUBJECT_CONFIG_DEFAULT, Importance.HIGH, SCHEMA_SUBJECT_CONFIG_DOC)
            //定义    部分              第一部分默认设置                    部分文件
            .define(CUID, Type.STRING, CUID_DEFAULT, Importance.MEDIUM, CUID_DOC);

    private String topic;
    private String stockPortfolio;
    private String refreshInterval;
    private String schemaUri;
    private String schemaSubject;
    private String spoofFlag;
    private String cuid;
    private String stockSymbols;

    /**
     * 读取 客户配置的信息
     * @param props
     * @author 代霞
     */
    @Override
    public void start(Map<String, String> props) {

        //从用户配置中读取信息
        topic = props.get(TOPIC_CONFIG);
        stockPortfolio = props.get(STOCK_PORTFOLIO_CONFIG);
        refreshInterval = props.get(REFRESH_INTERVAL_CONFIG);
        schemaUri = props.get(SCHEMA_URI_CONFIG);
        schemaSubject = props.get(SCHEMA_SUBJECT_CONFIG);
        spoofFlag = props.get(SPOOF_FLAG_CONFIG);
        cuid = props.get(CUID);

        //配置信息默认值设置
        if (topic == null)
            topic = "topic_stock";
        //传入不规范，则提示异常
        if (topic.contains(","))
            throw new ConnectException("股票读取连接器只支持一个主体topic");

        if(!stockPortfolio.equalsIgnoreCase("none")) {
            stockSymbols = YahooFinanceStockHelper.portfolio.get(stockPortfolio);
        } else {
            stockSymbols = props.get(STOCK_SYMBOLS_CONFIG);
            if(stockSymbols == null) stockSymbols = "INTC,TSLA";
        }

        if (refreshInterval != null && !refreshInterval.isEmpty()) {
            try {
                Integer.parseInt(refreshInterval);
            } catch (NumberFormatException nfe) {
                throw new ConnectException("'interval' must be a valid integer");
            }
        } else {
            refreshInterval = "20";
        }
        if(schemaUri==null)
            schemaUri="";
        if (schemaUri.endsWith("/"))
            schemaUri = schemaUri.substring(0, schemaUri.length() - 1);

        if (schemaSubject == null || schemaSubject.equalsIgnoreCase("n/a"))
            schemaSubject = topic;


    }

    /**
     *  加载 connector 的 task 类对象
     * @return
     */
    @Override
    public Class<? extends Task> taskClass() {

        return FinanceSourceTask.class;
    }

    /**
     *  根据 start 中读取新 生成 task 配置新
     * @param i
     * @return
     * @author 张栋
     */
    @Override
    // 创建一个名为taskConfigs类型为Map的List并将Start中生成的配置参数上传
    public List<Map<String, String>> taskConfigs(int i) {
        Map<String, String> config = new HashMap<String, String>();
        config.put(TOPIC_CONFIG, topic);
        config.put(STOCK_SYMBOLS_CONFIG, stockSymbols);
        config.put(STOCK_PORTFOLIO_CONFIG, stockPortfolio);
        config.put(REFRESH_INTERVAL_CONFIG, refreshInterval);
        config.put(SPOOF_FLAG_CONFIG, spoofFlag);
        config.put(SCHEMA_URI_CONFIG, schemaUri);
        config.put(SCHEMA_SUBJECT_CONFIG, schemaSubject);
        config.put(CUID, cuid);
        log.info("FinanceSourceConnector value: {}", getValues(config));
        return Arrays.asList(config);
    }

    private String getValues(Map<String, String> config) {
        StringBuilder builder = new StringBuilder();
        for (String key : config.keySet()) {
            builder.append("\n\t").append(key).append(" = ").append(config.get(key));
        }
        return builder.append("\n").toString();
    }

    /**
     * 停止
     */
    @Override
    public void stop() {

    }

    /**
     * 根据start 中的 创建 connector 配置对象
     * @return
     * @author 任士豪
     */
    @Override
    public ConfigDef config() {
        return CONFIG_DEF;
    }

    /**
     * 获取版本
     * @return
     */
    @Override
    public String version() {
        return "1.0-SNAPSHOT";
    }
}
